package drds.global_transaction.api.rpc;

import com.alibaba.fastjson.JSON;
import drds.common.$;
import drds.common.Constants;
import drds.common.Json;
import drds.common.Log;
import drds.global_transaction.api.message_codec.*;
import drds.global_transaction.api.rpc.rpc_message.RpcInfo;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import lombok.extern.slf4j.Slf4j;

import java.nio.ByteBuffer;
import java.text.MessageFormat;
import java.util.List;

@Slf4j
public class ByteAndRpcInfoCodec extends ByteToMessageCodec<RpcInfo>
{
	private static final int flag_async = 0x10;
	private static final int flag_request = 0x20;
	private static final int flag_heartbeat = 0x40;
	private static final int flag_other = 0x80;

	//
	private static final int not_found_index = -1;
	private static int head_length = 14;// 2<magic>+2<flag>+2<typeCode>+8<id>

	@Override
	protected void encode(ChannelHandlerContext channelHandlerContext, RpcInfo rpcInfo, ByteBuf byteBuf) throws Exception
	{
		IMessageCodec messageCodec = null;
		ByteBuffer byteBuffer = ByteBuffer.allocate(128);
		if (rpcInfo.getMessage() instanceof IMessageCodec)
		{
			messageCodec = (IMessageCodec) rpcInfo.getMessage();
		}
		byteBuffer.putShort(Magic.magic);
		int flag = (rpcInfo.isSynchronous() ? flag_async : 0) | //
				(rpcInfo.isRequest() ? flag_request : 0) | //
				(rpcInfo.isHeartbeat() ? flag_heartbeat : 0) | //
				flag_other;// keep old byte//
		byteBuffer.putShort((short) flag);
		//
		// byteBuffer to netty byteBuf
		if (rpcInfo.getMessage() instanceof HeartbeatMessage)
		{
			byteBuffer.putShort((short) 0);
			byteBuffer.putLong(rpcInfo.getId());
			byteBuffer.flip();
			//
			byte[] bytes = new byte[byteBuffer.limit()];
			byteBuffer.get(bytes);// 写入数组中
			// netty write
			byteBuf.writeBytes(bytes);
			return;
		}

		try
		{
			if (null != messageCodec)
			{
				byteBuffer.putShort(messageCodec.getTypeCode());
				byteBuffer.putLong(rpcInfo.getId());
				byteBuffer.flip();
				//
				byte[] bytes = new byte[byteBuffer.limit()];
				byteBuffer.get(bytes);
				byteBuf.writeBytes(bytes);// 写入数组中
				// netty write
				byte[] messageBytes = messageCodec.encode();
				if (Constants.developMode)
				{
					int length = messageBytes.length;
					Log.debugWithLine(log, Thread.currentThread(), "messageBytes size:" + length);
				}
				byteBuf.writeBytes(messageBytes);
			} else
			{
				throw new IllegalStateException();
			}
		} catch (Exception e)
		{
			log.error(rpcInfo.getMessage() + " encode error", "", e);
			throw e;
		}
	}

	/**
	 * @param byteBuf    输入
	 * @param objectList 输出
	 */
	@Override
	protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> objectList) throws Exception
	{
		int begin = byteBuf.readerIndex();
		int magicStartIndex = getMagicStartIndex(byteBuf);
		if (magicStartIndex == not_found_index)
		{
			log.error("magicStartIndex is not found");
			byteBuf.skipBytes(byteBuf.readableBytes());
			return;
		}
		byteBuf.skipBytes(magicStartIndex - byteBuf.readerIndex());
		if (byteBuf.readableBytes() < head_length)
		{
			log.error("decode less than header length");
			return;
		}
		byte[] bytes = new byte[head_length];
		byteBuf.readBytes(bytes);
		// netty byteBuf to byteBuffer
		ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
		byteBuffer.getShort();// magic
		//
		int flag = byteBuffer.getShort();
		if (Constants.developMode)
		{
			log.debug("flag:" + flag);
		}
		boolean async = (flag_async & flag) > 0;
		boolean isRequest = (flag_request & flag) > 0;
		boolean heartbeat = (flag_heartbeat & flag) > 0;
		//
		short typeCode = byteBuffer.getShort();
		long messageId = byteBuffer.getLong();
		if (heartbeat)
		{
			RpcInfo rpcInfo = new RpcInfo();
			rpcInfo.setId(messageId);
			rpcInfo.setSynchronous(true);
			rpcInfo.setRequest(isRequest);
			rpcInfo.setHeartbeat(true);
			if (isRequest)
			{
				rpcInfo.setMessage(HeartbeatMessage.ping);
			} else
			{
				rpcInfo.setMessage(HeartbeatMessage.pong);
			}
			objectList.add(rpcInfo);
			return;
		}
		RpcInfo rpcInfo = new RpcInfo();
		rpcInfo.setId(messageId);
		rpcInfo.setSynchronous(async);
		rpcInfo.setRequest(isRequest);
		rpcInfo.setHeartbeat(false);

		try
		{
			IMessageCodec messageCodec = MessageCodecs.getMessageCodec(typeCode);
			if (!messageCodec.decode(byteBuf))
			{
				log.error(messageCodec + " decode error.");
				byteBuf.readerIndex(begin);
				return;
			}
			if (Constants.developMode)
			{
				boolean request = messageCodec instanceof RequestMessage || messageCodec instanceof RequestMessages;
				String messageType = request ? "请求" : "响应";
				String message = MessageFormat.format("收到{0}:{1},信息内容是:{2}", messageType, messageCodec.getClass().getSimpleName(), JSON.toJSONString(messageCodec));
				Log.debugWithLine(log, Thread.currentThread(), message);
				String messageCodecBody = Json.formate(JSON.toJSONString(messageCodec));
				Log.output(log, messageCodecBody);
			}
			rpcInfo.setMessage(messageCodec);
		} catch (Exception e)
		{
			if (Constants.developMode)
			{
				e.printStackTrace();
			}
			log.error($.printStackTraceToString(e));
			//
			log.error("decode error", "", e);
			throw e;
		}
		objectList.add(rpcInfo);

	}

	private static int getMagicStartIndex(ByteBuf byteBuf)
	{
		boolean found = false;
		int index = byteBuf.readerIndex();
		int magicStartIndex = 0;
		while (index < byteBuf.writerIndex())
		{
			if (byteBuf.getByte(index) == Magic.magic_half && byteBuf.getByte(index + 1) == Magic.magic_half)
			{
				magicStartIndex = index;
				found = true;
				break;
			}
			++index;
		}
		return found ? magicStartIndex : not_found_index;
	}

}
